package com.atbeijing.D02;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * union
 * 合并同类型(泛型相同)的多个DataStream
 */
public class Example7 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> stream = env.fromElements("hello word");

        DataStreamSource<SensorReading> stream1 = env.addSource(new SensorSource());

        SingleOutputStreamOperator<SensorReading> sensor_1 = env.addSource(new SensorSource()).filter(r -> r.id.equals("sensor_1"));
        SingleOutputStreamOperator<SensorReading> sensor_2 = env.addSource(new SensorSource()).filter(r -> r.id.equals("sensor_2"));
        SingleOutputStreamOperator<SensorReading> sensor_3 = env.addSource(new SensorSource()).filter(r -> r.id.equals("sensor_3"));

        //不同类型的流不能union,泛型不同
        //stream.union(sensor_1);错误
        //stream.union(stream1);错误

        //合并流
        DataStream<SensorReading> union1 = stream1.union(sensor_1);
        DataStream<SensorReading> union = sensor_1.union(sensor_2, sensor_3);
        env.execute();
    }
}
